package com.dubbo.rpc;

import com.dubbo.rpc.codec.RpcDecoder;
import com.dubbo.rpc.codec.RpcEncoder;
import com.dubbo.rpc.discovery.ServiceDiscovery;
import com.dubbo.rpc.domain.RpcRequest;
import com.dubbo.rpc.domain.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.Proxy;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author liuxin
 * @date 2021/10/10 9:43
 */
@Component
public class RpcClient {

    @Resource
    private ServiceDiscovery serviceDiscovery;

    private ConcurrentHashMap<String, RpcResponse> responseMap = new ConcurrentHashMap<>();


    /**
     * 每次调用都会调用动态代理, 并发量大时考虑性能问题
     * @param interfaceCls
     * @param <T>
     * @return
     */
    public <T> T create(final Class<?> interfaceCls) {
        return (T) Proxy.newProxyInstance(
                interfaceCls.getClassLoader(),
                new Class<?>[]{interfaceCls},
                (proxy, method, args) -> {
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setInterfaceName(method.getDeclaringClass().getName());
                    rpcRequest.setMethodName(method.getName());
                    rpcRequest.setParamsTypeArr(method.getParameterTypes());
                    rpcRequest.setParamsValArr(args);

                    String serviceName = interfaceCls.getName();
                    String serviceAddress = serviceDiscovery.discover(serviceName);
                    if (StringUtils.isBlank(serviceAddress)) {
                        throw new RuntimeException("serviceAddress blank");
                    }

                    String[] array = StringUtils.split(serviceAddress, ":");
                    String host = array[0];
                    int port = NumberUtils.toInt(array[1]);

                    RpcResponse rpcResponse = send(rpcRequest, host, port);
                    if (Objects.isNull(rpcResponse)) {
                        return null;
                    }

                    return rpcResponse.getData();
                }
        );
    }

    private RpcResponse send(RpcRequest request, String host, int port) {
        NioEventLoopGroup group = new NioEventLoopGroup(1);

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();

                    pipeline.addLast(new RpcEncoder(RpcRequest.class));
                    pipeline.addLast(new RpcDecoder(RpcResponse.class));
                    pipeline.addLast(new RpcClientHandler(responseMap));
                }
            });

            ChannelFuture future = bootstrap.connect(host, port).sync();
            Channel channel = future.channel();

            channel.writeAndFlush(request).sync();
            channel.closeFuture().sync();

            return responseMap.get(request.getRequestId());
        } catch (Exception e) {
            System.err.println(ExceptionUtils.getStackTrace(e));
            return null;
        } finally {
            group.shutdownGracefully();
            responseMap.remove(request.getRequestId());
        }
    }

}
